package com.yunhe.abnormal.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.abnormal.client.AuthorityClient;
import com.yunhe.abnormal.client.DeviceClient;
import com.yunhe.common.constant.AbnormalConsts;
import com.yunhe.common.constant.DataTransferConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.SystemConsts;
import com.yunhe.abnormal.consts.AlarmConsts;
import com.yunhe.abnormal.domain.base.AbnormalRule;
import com.yunhe.abnormal.domain.base.AlarmLevel;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.exception.ObjectNotFoundException;
import com.yunhe.common.exception.UniqueConstraintsException;
import com.yunhe.common.model.authority.FirmDTO;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.common.model.authority.UserDTO;
import com.yunhe.common.model.base.*;
import com.yunhe.common.model.data.AbnormalRuleFormulaObject;
import com.yunhe.common.model.device.Analog;
import com.yunhe.common.model.device.DeviceDTO;
import com.yunhe.common.model.msg.*;
import com.yunhe.abnormal.service.base.*;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.StringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import zju.fes.exp.IllegalExpressionException;
import zju.fes.exp.IncalculableExpressionException;
import zju.fes.exp.UnknownOperatorException;

import java.io.IOException;
import java.util.*;

/**
 * 告警规则生成监听
 */
@Component
public class AlarmRuleListener {
    private static final Logger logger = LoggerFactory.getLogger(AlarmRuleListener.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();

    @Autowired
    private AlarmLevelService alarmLevelService;
    @Autowired
    private AlarmRuleService alarmRuleService;
    @Autowired
    private AlarmNotifyTypeService alarmNotifyTypeService;
    @Autowired
    private AlarmRuleInstanceService alarmRuleInstanceService;
    @Autowired
    private AbnormalRuleService abnormalRuleService;
    @Autowired
    private AlarmNotifyConfigService alarmNotifyConfigService;
    @Autowired
    private DeviceClient deviceClient;
    @Autowired
    private AuthorityClient authorityClient;

    /**
     * 设备+数据项与点号Map对应的Key
     * @param deviceId 设备ID/电站ID
     * @param typeId 数据项类型
     */
    private String getTypePointMapKey(Long deviceId, Long typeId) {
        return deviceId + "," + typeId;
    }

    /**
     * 设备+数据项与点号的映射
     * @param stations 电站列表
     * @param devices 设备列表
     * @param pointDataTypes 数据项类型
     * @param relatedPointDataTypes 关联数据项类型
     */
    private Map<String, Long> getTypePointMap(List<StationDTO> stations,
                                              List<DeviceDTO> devices,
                                              List<PointDataTypeDTO> pointDataTypes,
                                              List<PointDataTypeDTO> relatedPointDataTypes) {
        Map<String, Long> typePointMap = new HashMap<>();
        List<Long> devIds = new ArrayList<>();
        if (stations != null) {
            for (StationDTO station : stations) {
                devIds.add(station.getId());
            }
        }
        if (devices != null) {
            for (DeviceDTO device : devices) {
                devIds.add(device.getId());
            }
        }
        List<Long> typeIds = new ArrayList<>();
        if (pointDataTypes != null) {
            for (PointDataTypeDTO pointDataType : pointDataTypes) {
                typeIds.add(pointDataType.getId());
            }
        }
        if (relatedPointDataTypes != null) {
            for (PointDataTypeDTO pointDataType : relatedPointDataTypes) {
                typeIds.add(pointDataType.getId());
            }
        }
        List<Analog> analogs = deviceClient.getAnalogsByDevAndType(devIds, typeIds, null).getBody().getResults();
        if (analogs != null) {
            for (Analog analog : analogs) {
                typePointMap.put(getTypePointMapKey(analog.getDeviceId(), analog.getTypeId()), analog.getPointNumber());
            }
        }
        return typePointMap;
    }

    /**
     * 创建异常规则实例
     * @param alarmRule 异常规则
     * @param typePointMap 设备+数据项与点号的映射
     * @param station 电站
     * @param device 设备
     */
    private AlarmRuleInstanceDTO createInstance(AlarmRuleDTO alarmRule, Map<String, Long> typePointMap, StationDTO station, DeviceDTO device) {
        if (station == null && device == null) {
            return null;
        }
        String formula = alarmRule.getFormula();
        List<PointDataTypeDTO> pointDataTypes = alarmRule.getPointDataTypes();
        List<PointDataTypeDTO> relatedPointDataTypes = alarmRule.getRelatedPointDataTypes();
        List<Long> pointDataTypeIds = new ArrayList<>();
        List<Long> relatedPointDataTypeIds = new ArrayList<>();
        if (pointDataTypes != null) {
            for (PointDataTypeDTO pointDataType : pointDataTypes) {
                Long pointNumber = typePointMap.get(getTypePointMapKey(station != null ? station.getId() : device.getId(), pointDataType.getId()));
                if (pointNumber != null) {
                    pointDataTypeIds.add(pointNumber);
                    formula = formula.replaceAll(pointDataType.getTitle(), String.valueOf(pointNumber));
                } else {
                    // 如果找不到公式中的点号，则不添加公式
                    return null;
                }
            }
        }
        if (relatedPointDataTypes != null) {
            for (PointDataTypeDTO pointDataType : relatedPointDataTypes) {
                Long pointNumber = typePointMap.get(getTypePointMapKey(station != null ? station.getId() : device.getId(), pointDataType.getId()));
                if (pointNumber != null) {
                    relatedPointDataTypeIds.add(pointNumber);
                }
            }
        }
        AlarmRuleInstanceDTO alarmRuleInstance = new AlarmRuleInstanceDTO();
        alarmRuleInstance.setAlarmRule(alarmRule);
        if (station != null) {
            // 设置适用电站
            alarmRuleInstance.setStation(station);
            alarmRuleInstance.setStationId(station.getId());
        } else {
            // 设置适用设备
            alarmRuleInstance.setDevice(device);
            // 设置所属电站ID
            StationDTO bindStation = authorityClient.getStationByDeviceId(device.getId()).getBody().getResults();
            if (bindStation != null) {
                alarmRuleInstance.setStationId(bindStation.getId());
            }
        }
        alarmRuleInstance.setFormula(formula);
        alarmRuleInstance.setPoints(pointDataTypeIds);
        alarmRuleInstance.setRelatedPoints(relatedPointDataTypeIds);
        return alarmRuleInstance;
    }

    /**
     * 异常规则实例化
     * @param alarmRule 异常规则
     * @param stations 适用电站
     * @param devices 适用设备
     */
    private List<AlarmRuleInstanceDTO> getRuleInstances(AlarmRuleDTO alarmRule,
                                                        List<StationDTO> stations,
                                                        List<DeviceDTO> devices) {
        List<AlarmRuleInstanceDTO> result = new ArrayList<>();
        if (stations == null && alarmRule.getStationType() != null) {
            if (alarmRule.getIsAllStation()) {
                stations = authorityClient.searchStation(alarmRule.getStationType().getId()).getBody().getResults();
            } else {
                stations = alarmRule.getStations();
            }
        }
        if (devices == null && alarmRule.getDeviceType() != null) {
            if (alarmRule.getIsAllDevice()) {
                devices = deviceClient.searchDevice(alarmRule.getDeviceType().getId()).getBody().getResults();
            } else {
                devices = alarmRule.getDevices();
            }
        }
        List<PointDataTypeDTO> pointDataTypes = alarmRule.getPointDataTypes();
        List<PointDataTypeDTO> relatedPointDataTypes = alarmRule.getRelatedPointDataTypes();
        Map<String, Long> typePointMap = getTypePointMap(stations, devices, pointDataTypes, relatedPointDataTypes);
        if (devices != null) {
            for (DeviceDTO device : devices) {
                AlarmRuleInstanceDTO instance = createInstance(alarmRule, typePointMap, null, device);
                if (instance != null) {
                    result.add(instance);
                }
            }
        }
        if (stations != null) {
            for (StationDTO station : stations) {
                AlarmRuleInstanceDTO instance = createInstance(alarmRule, typePointMap, station, null);
                if (instance != null) {
                    result.add(instance);
                }
            }
        }
        return result;
    }

    /**
     * 用户新增/删除时对异常通知配置进行更新
     * @param kafkaRecord 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.UserRefresh, groupId = KafkaConsts.Group.AlarmNotifyConfigProcess)
    public void userRefreshListener(ConsumerRecord<?, String> kafkaRecord) throws ObjectNotFoundException, IOException {
        try {
            String value = kafkaRecord.value();
            if (!StringUtil.isEmpty(value)) {
                UserMsgDTO msg = mapper.readValue(value, UserMsgDTO.class);
                String action = msg.getAction();
                List<UserDTO> users = msg.getUsers();
                if (SystemConsts.ACTION.ADD.equals(action)) {
                    List<AlarmNotifyTypeDTO> alarmNotifyTypes = alarmNotifyTypeService.findAll();
                    List<AlarmLevelDTO> alarmLevels = alarmLevelService.findAll();
                    List<AlarmNotifyConfigDTO> newConfig = new ArrayList<>();
                    for (UserDTO user : users) {
                        Long sn = 1L;
                        for (AlarmLevelDTO alarmLevel : alarmLevels) {
                            for (AlarmNotifyTypeDTO alarmNotifyType : alarmNotifyTypes) {
                                AlarmNotifyConfigDTO alarmNotifyConfig = new AlarmNotifyConfigDTO();
                                alarmNotifyConfig.setAlarmLevel(alarmLevel);
                                alarmNotifyConfig.setAlarmNotifyType(alarmNotifyType);
                                alarmNotifyConfig.setUser(user);
                                alarmNotifyConfig.setSn(sn++);
                                alarmNotifyConfig.setType(AlarmConsts.AlarmNotifyConfigType.USER.value());
                                alarmNotifyConfig.setStatus(AlarmConsts.AlarmNotifyStatus.OPENED.value());
                                newConfig.add(alarmNotifyConfig);
                            }
                        }
                    }
                    alarmNotifyConfigService.create(newConfig);
                } else if (SystemConsts.ACTION.DELETE.equals(action)) {
                    for (UserDTO user : users) {
                        alarmNotifyConfigService.deleteByUserId(user.getId());
                    }
                }
            }
        } catch (Exception e) {
            logger.error("新增告警APP推送报错，{}，{}", kafkaRecord.value(), e);
        }
    }

    /**
     * 单位新增/删除时对异常通知配置进行更新
     * @param kafkaRecord 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.FirmRefresh, groupId = KafkaConsts.Group.AlarmNotifyConfigProcess)
    public void firmRefreshListener(ConsumerRecord<?, String> kafkaRecord) throws ObjectNotFoundException, IOException {
        try {
            String value = kafkaRecord.value();
            if (!StringUtil.isEmpty(value)) {
                FirmMsgDTO msg = mapper.readValue(value, FirmMsgDTO.class);
                String action = msg.getAction();
                List<FirmDTO> firms = msg.getFirms();
                if (SystemConsts.ACTION.ADD.equals(action)) {
                    List<AlarmNotifyTypeDTO> alarmNotifyTypes = alarmNotifyTypeService.findAll();
                    List<AlarmLevelDTO> alarmLevels = alarmLevelService.findAll();
                    List<AlarmNotifyConfigDTO> newConfig = new ArrayList<>();
                    for (FirmDTO firm : firms) {
                        Long sn = 1L;
                        for (AlarmLevelDTO alarmLevel : alarmLevels) {
                            for (AlarmNotifyTypeDTO alarmNotifyType : alarmNotifyTypes) {
                                AlarmNotifyConfigDTO alarmNotifyConfig = new AlarmNotifyConfigDTO();
                                alarmNotifyConfig.setAlarmLevel(alarmLevel);
                                alarmNotifyConfig.setAlarmNotifyType(alarmNotifyType);
                                alarmNotifyConfig.setFirm(firm);
                                alarmNotifyConfig.setSn(sn++);
                                alarmNotifyConfig.setType(AlarmConsts.AlarmNotifyConfigType.FIRM.value());
                                alarmNotifyConfig.setStatus(AlarmConsts.AlarmNotifyStatus.OPENED.value());
                                newConfig.add(alarmNotifyConfig);
                            }
                        }
                    }
                    alarmNotifyConfigService.create(newConfig);
                } else if (SystemConsts.ACTION.DELETE.equals(action)) {
                    for (FirmDTO firm : firms) {
                        alarmNotifyConfigService.deleteByFirmId(firm.getId());
                    }
                }
            }
        } catch (Exception e) {
            logger.error("新增告警APP推送报错，{}，{}", kafkaRecord.value(), e);
        }
    }

    /**
     * 新增/删除电站时初始化异常规则实例
     * @param kafkaRecord 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.StationRefresh, groupId = KafkaConsts.Group.AlarmRuleInstanceProcess)
    public void stationRefreshListener(ConsumerRecord<?, String> kafkaRecord) throws ObjectNotFoundException, IOException {
        try {
            String value = kafkaRecord.value();
            if (!StringUtil.isEmpty(value)) {
                StationMsgDTO msg = mapper.readValue(value, StationMsgDTO.class);
                String action = msg.getAction();
                List<StationDTO> stations = msg.getStations();
                if (SystemConsts.ACTION.ADD.equals(action)) {
//                    for (StationDTO station : stations) {
//                        List<AlarmRuleDTO> alarmRules = alarmRuleService.findAlarmRuleForStation(station.getId());
//                        if (CollectionUtil.isNotEmpty(alarmRules)) {
//                            List<AlarmRuleInstanceDTO> alarmRuleInstances = getRuleInstances(alarmRules,
//                                    Collections.singletonList(station), null);
//                            if (alarmRuleInstances.size() > 0) {
//                                alarmRuleInstanceService.create(alarmRuleInstances);
//                            }
//                        }
//                    }
                } else if (SystemConsts.ACTION.DELETE.equals(action)) {
                    List<Long> stationIds = new ArrayList<>();
                    for (StationDTO station : stations) {
                        stationIds.add(station.getId());
                    }
                    alarmRuleInstanceService.deleteByStationId(stationIds);
                }
            }
        } catch (Exception e) {
            logger.error("新增/删除电站时初始化异常规则实例报错，{}，{}", kafkaRecord.value(), e);
        }
    }

    /**
     * 新增/删除设备时初始化异常规则实例
     * @param kafkaRecord 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.DeviceRefresh, groupId = KafkaConsts.Group.AlarmRuleInstanceProcess)
    public void deviceRefreshListener(ConsumerRecord<?, String> kafkaRecord) throws ObjectNotFoundException, IOException {
        try {
            String value = kafkaRecord.value();
            if (!StringUtil.isEmpty(value)) {
                DeviceMsgDTO msg = mapper.readValue(value, DeviceMsgDTO.class);
                String action = msg.getAction();
                List<DeviceDTO> devices = msg.getDevices();
                if (SystemConsts.ACTION.ADD.equals(action)) {
//                    for (DeviceDTO device : devices) {
//                        List<AlarmRuleDTO> alarmRules = alarmRuleService.findAlarmRuleForDevice(device.getId());
//                        if (CollectionUtil.isNotEmpty(alarmRules)) {
//                            List<AlarmRuleInstanceDTO> alarmRuleInstances = getRuleInstances(alarmRules,
//                                    null, Collections.singletonList(device));
//                            if (alarmRuleInstances.size() > 0) {
//                                alarmRuleInstanceService.create(alarmRuleInstances);
//                            }
//                        }
//                    }
                } else if (SystemConsts.ACTION.DELETE.equals(action)) {
                    List<Long> deviceIds = new ArrayList<>();
                    for (DeviceDTO deviceId : devices) {
                        deviceIds.add(deviceId.getId());
                    }
                    alarmRuleInstanceService.deleteByDeviceId(deviceIds);
                }
            }
        } catch (Exception e) {
            logger.error("新增/删除设备时初始化异常规则实例报错，{}，{}", kafkaRecord.value(), e);
        }
    }

    /**
     * 新增/删除/修改异常规则时初始化异常规则实例
     * @param kafkaRecord 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmRuleRefresh, groupId = KafkaConsts.Group.AlarmRuleInstanceProcess)
    public void alarmRuleRefreshListener(ConsumerRecord<?, String> kafkaRecord) throws ObjectNotFoundException, IOException {
        try {
            String value = kafkaRecord.value();
            if (!StringUtil.isEmpty(value)) {
                AlarmRuleMsgDTO msg = mapper.readValue(value, AlarmRuleMsgDTO.class);
                String action = msg.getAction();
                List<AlarmRuleDTO> alarmRules = msg.getAlarmRules();
                if (SystemConsts.ACTION.ADD.equals(action)) {
                    List<AlarmRuleInstanceDTO> alarmRuleInstances = new ArrayList<>();
                    for (AlarmRuleDTO alarmRule : alarmRules) {
                        alarmRuleInstances.addAll(getRuleInstances(alarmRule, null, null));
                    }
                    if (alarmRuleInstances.size() > 0) {
                        alarmRuleInstanceService.create(alarmRuleInstances);
                    }
                } else if (SystemConsts.ACTION.DELETE.equals(action)) {
                    List<Long> alarmRuleIds = new ArrayList<>();
                    for (AlarmRuleDTO alarmRule : alarmRules) {
                        alarmRuleIds.add(alarmRule.getId());
                    }
                    alarmRuleInstanceService.deleteByRuleId(alarmRuleIds);
                } else if (SystemConsts.ACTION.EDIT.equals(action)) {
                    for (AlarmRuleDTO alarmRule : alarmRules) {
                        List<AlarmRuleInstanceDTO> alarmRuleInstances = getRuleInstances(alarmRule, null, null);
                        alarmRuleInstanceService.updateByRule(alarmRule.getId(), alarmRuleInstances);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("新增/删除/修改异常规则时初始化异常规则实例报错，{}，{}", kafkaRecord.value(), e);
        }
    }

    /**
     * 新增电站时添加心跳告警规则
     * 触发方式：电站新增
     * @param record 消息
     */
    @KafkaListener(topics = KafkaConsts.Topic.StationRefresh, groupId = KafkaConsts.Group.StationHeartBeatAlarmRule)
    public void generateStationAnalogs(ConsumerRecord<?, String> record) {
        try {
            String value = record.value();
            if (!StringUtil.isEmpty(value)) {
                StationMsgDTO msg = mapper.readValue(value, StationMsgDTO.class);
                String action = msg.getAction();
                if (SystemConsts.ACTION.ADD.equals(action)) {
                    List<StationDTO> stations = msg.getStations();
                    for (StationDTO station : stations) {
                        // TODO 先注释掉，暂时不生成告警公式
                        //addHeartbeatRule(station);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("新增电站时添加心跳告警规则报错，{}，{}", record.value(), e);
        }
    }

    /**
     * 新增电站时添加心跳告警规则
     * 触发方式：电站新增
     * @param station 电站对象
     */
    private void addHeartbeatRule(StationDTO station) throws IOException, ArgumentErrorException, UniqueConstraintsException,
        ObjectNotFoundException, IncalculableExpressionException, UnknownOperatorException, IllegalExpressionException {
        Long stationId = station.getId();
        Long stationCode = station.getCode();
        //新增电站离线告警
        if (stationCode == null) {
            throw new ArgumentErrorException("电站代码不能为空");
        }
        Long heartbeatPoint = DataTransferConsts.getHeartbeatPid(stationCode);
        AlarmLevel serious = alarmLevelService.findByName(AbnormalConsts.ALARM_LEVEL.SERIOUS.value());
        AbnormalRule heartBeatRule = new AbnormalRule();
        heartBeatRule.setObjectId(stationId);
        if ("Asia/Shanghai".equals(station.getTimeZone()) || station.getTimeZone() == null) {
            heartBeatRule.setTitle(AbnormalConsts.ALARM_RULE_TITLE.HEARTBEAT_CN.getValue());
        } else {
            heartBeatRule.setTitle(AbnormalConsts.ALARM_RULE_TITLE.HEARTBEAT_EN.getValue());
        }
        AbnormalRuleFormulaObject formulaObject = new AbnormalRuleFormulaObject();
        formulaObject.setFormula(generateHeartBeatFormula(String.valueOf(heartbeatPoint)));
        formulaObject.setMeasPointNumbers(Collections.singletonList(heartbeatPoint));
        formulaObject.setAlarmLevelId(serious.getId());
        heartBeatRule.setAbnormalRuleFormulaObjects(Collections.singletonList(formulaObject));
        abnormalRuleService.create(heartBeatRule);
    }

    /**
     * 生成告警公式
     * @param pointNumber 点号
     */
    private String generateHeartBeatFormula(String pointNumber) {
        return "var(" + pointNumber + ")=1";
    }
}
